package jmeter.io;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import bot.Hand;

public class EchoIOServer {
	protected static final Logger LOCAL_LOG = LogManager
			.getLogger(LogManager.ROOT_LOGGER_NAME);

	private final int port;

	private final BlockingQueue<SocketData> receivedSockets = new LinkedBlockingQueue<SocketData>();

	private final ExecutorService executorServiceRead = Executors
			.newFixedThreadPool(7);

	private final ExecutorService executorServiceWrite = Executors
			.newFixedThreadPool(7);

	private final ServerSocket serverSocket;

	private final ThreadWrite threadWrite = new ThreadWrite(
			getReceivedSockets());

	private final Hand hand = new Hand();

	public EchoIOServer(int port) throws IOException {
		this.port = port;
		serverSocket = new ServerSocket();
	}

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		EchoIOServer ioServer = null;
		Exception exception = null;
		try {
			ioServer = new EchoIOServer(12345);

			ioServer.serverSocket.bind(new InetSocketAddress("localhost",
					ioServer.port));

			while (true)
				try {
					ioServer.executorServiceRead
							.submit(ioServer.new ThreadRead(
									ioServer.serverSocket.accept(), ioServer
											.getReceivedSockets()));
				} catch (IOException ioException) {
					LOCAL_LOG.error(ioException.getMessage(), ioException);
				} catch (Throwable t) {
					LOCAL_LOG.fatal(t.getMessage(), t);
				}
		} catch (IOException e) {
			exception = e;
		} finally {
			if (exception != null)
				LOCAL_LOG.error(exception.toString(), exception);
			if (ioServer != null)
				if (ioServer.serverSocket != null)
					try {
						ioServer.serverSocket.close();
					} catch (IOException e) {
						LOCAL_LOG.error(e.getMessage(), e);
					}
			ioServer.executorServiceRead.shutdown();
			LOCAL_LOG.debug("Accepting thread pool shutdown.");
		}
	}

	public BlockingQueue<SocketData> getReceivedSockets() {
		return receivedSockets;
	}

	public ThreadWrite getThreadWrite() {
		return threadWrite;
	}

	private class ThreadRead implements Runnable {
		private final Socket sCurrent;
		private final BlockingQueue<SocketData> out;

		public ThreadRead(final Socket socket,
				final BlockingQueue<SocketData> blockingQueue) {
			this.sCurrent = socket;
			this.out = blockingQueue;
		}

		@Override
		public void run() {
			int count = -1;
			// do {
			try {
				byte[] cash = new byte[Byte.MAX_VALUE];
				count = sCurrent.getInputStream().read(cash);

				byte[] bytes = new byte[count];
				for (int i = 0; i < count; i++)
					bytes[i] = cash[i];

				Future<AtomicBoolean> future = executorServiceWrite
						.submit(hand);
				try {
					future.get();
				} catch (InterruptedException e) {
					LOCAL_LOG.error(e.getMessage(), e);
				} catch (ExecutionException e) {
					LOCAL_LOG.error(e.getMessage(), e);
				}
				SocketData data = new SocketData(sCurrent, bytes);
				executorServiceWrite.submit(getThreadWrite());
				out.offer(data);
				// sCurrent.getOutputStream().write(bytes);
				// sCurrent.close();
			} catch (IOException e) {
				LOCAL_LOG.error(e.getMessage(), e);
			}
			// } while (true);
		}
	}

	private class ThreadWrite implements Runnable {
		private final BlockingQueue<SocketData> out;
		private final SocketData[] datas = new SocketData[0];

		public ThreadWrite(final BlockingQueue<SocketData> blockingQueue) {
			this.out = blockingQueue;
		}

		@Override
		public void run() {
			SocketData sCurrent = null;
			try {
				sCurrent = this.out.poll();
				LOCAL_LOG.info(out.size());
				if (sCurrent != null) {
					byte[] b = sCurrent.getBs();
					sCurrent.getSocket().getOutputStream().write(b);
					for (SocketData s : out.toArray(datas)) {
						if (s != null)
							synchronized (s.getSocket()) {
								if (!s.getSocket().isClosed())
									s.getSocket().getOutputStream().write(b);
							}
					}
				}
			} catch (IOException e) {
				LOCAL_LOG.error(e.getMessage(), e);
			} finally {
				if (sCurrent != null)
					try {
						synchronized (sCurrent.getSocket()) {
							sCurrent.getSocket().close();
						}
					} catch (IOException e) {
						LOCAL_LOG.error(e.getMessage(), e);
					}
			}
		}
	}

	private class SocketData {
		private final Socket socket;
		private final byte[] bs;

		public SocketData(Socket socket, byte[] bs) {
			this.socket = socket;
			this.bs = bs;
		}

		public Socket getSocket() {
			return socket;
		}

		public byte[] getBs() {
			return bs;
		}
	}
}
